package org.finesys.common.hazelcast.core.mq;

import com.hazelcast.collection.IQueue;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.topic.ITopic;
import lombok.RequiredArgsConstructor;

import java.util.concurrent.ExecutorService;

@RequiredArgsConstructor
public class HazelcastMessageListenerProvider {

    private final HazelcastInstance hazelcastInstance;
    private final ExecutorService executorService;

    /**
     * Topic注册监听
     */
    public <M extends AbstractTopicMessage> void register(AbstractTopicMessageListener<M> listener) {
        String topicName = listener.getTopic();
        ITopic<M> topic = hazelcastInstance.getTopic(topicName);
        topic.addMessageListener(listener);
    }

    /**
     * 组消息的注册监听
     */
    public <M extends AbstractGroupMessage> void register(AbstractGroupMessageListener<M> listener) {
        String groupName = listener.getGroup();
        IQueue<M> queue = hazelcastInstance.getQueue(groupName);
        Thread worker = new Thread(() -> {
            while (true) {
                try {
                    M m = queue.take();
                    executorService.execute(() -> {
                        listener.onMessage(m);
                    });
                } catch (InterruptedException e) {

                }
            }
        }, "HazelcastConsumerThread-" + groupName);
        worker.start();
    }
}
